package com.zyht.redis.service.impl;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import com.wx.util.JsonUtil;
import com.zyht.redis.LPopBakAction;
import com.zyht.redis.RedisCash;
import com.zyht.redis.service.RedisMsgListener;
import com.zyht.redis.service.RedisService;

@Service
public class RedisServiceImpl implements RedisService {

	private static final Logger logger = LoggerFactory.getLogger(RedisServiceImpl.class);

	@Resource
	private RedisTemplate<Serializable, Serializable> redisTemplate;

	private byte[] getStringSerialize(String value) {
		return redisTemplate.getStringSerializer().serialize(value);
	}

	private String getStringDeserialize(byte[] value) {
		return redisTemplate.getStringSerializer().deserialize(value);
	}

	private byte[][] getStringSerializes(String... params) {
		byte[][] p = new byte[params.length][];
		for (int i = 0; i < params.length; i++)
			p[i] = getStringSerialize(params[i]);
		return p;
	}

	private List<String> getStringDeserialize(List<byte[]> bList) {
		List<String> sList = new ArrayList<String>();
		if (bList != null) {
			for (byte[] bs : bList) {
				sList.add(getStringDeserialize(bs));
			}
		}
		return sList;
	}

	private Map<String, String> getStringDeserialize(Map<byte[], byte[]> bMap) {
		Map<String, String> sMap = new HashMap<>();
		if (bMap != null) {
			for (Entry<byte[], byte[]> entry : bMap.entrySet()) {
				sMap.put(getStringDeserialize(entry.getKey()), getStringDeserialize(entry.getValue()));
			}
		}
		return sMap;
	}

	/**
	 * 将字符串值 value 关联到 key 。
	 * 
	 * @param key
	 * @param value
	 */
	@Override
	public void set(String key, String value) {
		set(getStringSerialize(key), getStringSerialize(value));
	}

	/**
	 * 将字符串值 value 关联到 key 。
	 * 
	 * @param key
	 * @param value
	 */
	@Override
	public void set(byte[] key, byte[] value) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) {
				connection.set(key, value);
				return null;
			}
		});
	}

	@Override
	public Boolean setNX(String key, String value) {
		return setNX(getStringSerialize(key), getStringSerialize(value));
	}

	@Override
	public Boolean setNX(byte[] key, byte[] value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.setNX(key, value);
			}
		});
	}

	/**
	 * 删除
	 * 
	 * @param keys
	 * @return
	 */
	@Override
	public Long del(String... keys) {
		return del(getStringSerializes(keys));
	}

	/**
	 * 删除
	 * 
	 * @param keys
	 * @return
	 */
	@Override
	public Long del(byte[]... keys) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.del(keys);
			}
		});
	}

	/**
	 * 获取key
	 */
	@Override
	public String get(String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) {
				return getStringDeserialize(connection.get(getStringSerialize(key)));
			}
		});
	}

	/**
	 * 获取key
	 */
	@Override
	public byte[] get(byte[] key) {
		return redisTemplate.execute(new RedisCallback<byte[]>() {
			@Override
			public byte[] doInRedis(RedisConnection connection) {
				return connection.get(key);
			}
		});
	}

	/**
	 * 执行脚本
	 * 
	 * @param script
	 * @param resultType
	 * @param value
	 * @return
	 */
	@Override
	public <T> T eval(String script, Class<T> resultType, String... value) {
		String scriptSha = scriptLoad(script);
		return evalSha(scriptSha, resultType, value);
	}

	/**
	 * 执行脚本
	 * 
	 * @param scriptSha
	 * @param resultType
	 * @param value
	 * @return
	 */
	@Override
	public <T> T evalSha(String scriptSha, Class<T> resultType, String... value) {
		return redisTemplate.execute(new RedisCallback<T>() {
			@Override
			public T doInRedis(RedisConnection connection) {
				ReturnType returnType = null;
				if (resultType == null) {
					returnType = ReturnType.STATUS;
				} else if (resultType.isAssignableFrom(String.class)) {
					returnType = ReturnType.STATUS;
				} else if (resultType.isAssignableFrom(List.class)) {
					returnType = ReturnType.MULTI;
				} else if (resultType.isAssignableFrom(Boolean.class)) {
					returnType = ReturnType.BOOLEAN;
				} else if (resultType.isAssignableFrom(Long.class)) {
					returnType = ReturnType.INTEGER;
				}
				return connection.evalSha(scriptSha, returnType, value.length, getStringSerializes(value));
			}
		});
	}

	/**
	 * list.add
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	@Override
	public Long lPush(String key, String value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.lPush(getStringSerialize(key), getStringSerialize(value));
			}
		});
	}

	@Override
	public Long lPush(String key, byte[] value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.lPush(getStringSerialize(key), value);
			}
		});
	}
	
	@Override
	public Long lPush(String key, String value, Long seconds) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				byte[] keyB = getStringSerialize(key);
				Long res = connection.lPush(keyB, getStringSerialize(value));
				if (seconds != null && seconds > 0)
					connection.expire(keyB, seconds);
				return res;
			}
		});
	}

	/**
	 * list.addall
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	@Override
	public Boolean expire(String key, Long seconds) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				byte[] keyB = getStringSerialize(key);
				return connection.expire(keyB, seconds);
			}
		});
	}
	
	@Override
	public Boolean expire(byte[] key, Long seconds) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.expire(key, seconds);
			}
		});
	}

	/**
	 * list.addall
	 * 
	 * @param key
	 * @param value
	 * @return
	 */
	@Override
	public Long lPush(String key, String... value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.lPush(getStringSerialize(key), getStringSerializes(value));
			}
		});
	}

	/**
	 * list.size
	 * 
	 * @param key
	 * @return
	 */
	@Override
	public Long lLen(String key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.lLen(getStringSerialize(key));
			}
		});
	}

	/**
	 * hExists
	 * 
	 * @param key
	 * @return
	 */
	@Override
	public Boolean hExists(String key, String value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.hExists(getStringSerialize(key), getStringSerialize(value));
			}
		});
	}

	/**
	 * 
	 * @param key
	 * @param field
	 * @param value
	 * @return
	 */
	@Override
	public Boolean hSet(String key, String field, String value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.hSet(getStringSerialize(key), getStringSerialize(field), getStringSerialize(value));
			}
		});
	}

	@Override
	public void hMSet(String key, Map<String, String> map) {
		Map<byte[], byte[]> hashes = new HashMap<byte[], byte[]>();
		if (map != null) {
			Set<Entry<String,String>> entrySet = map.entrySet();
			for (Entry<String, String> entry : entrySet) {
				if (entry.getKey() != null && entry.getValue() != null) {
					hashes.put(getStringSerialize(entry.getKey()), getStringSerialize(entry.getValue()));
				}
			}
			/*
			for (Iterator<Entry<String, String>> iterator = map.entrySet().iterator(); iterator.hasNext();) {
				Entry<String, String> type = iterator.next();
				if (type.getKey() != null && type.getValue() != null) {
					hashes.put(getStringSerialize(type.getKey()), getStringSerialize(type.getValue()));
				}
			}*/
		}
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) {
				connection.hMSet(getStringSerialize(key), hashes);
				return null;
			}
		});
	}

	@Override
	public Long hDel(String key, String... field) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.hDel(getStringSerialize(key), getStringSerializes(field));
			}
		});
	}

	/**
	 * 删除目标redis全部DB缓存
	 */
	@Override
	public void flushAll() {
		redisTemplate.execute(new RedisCallback<Object>() {
			public Object doInRedis(RedisConnection connection) {
				connection.flushAll();
				return null;
			}
		});
	}

	/**
	 * 预加载脚本
	 * 
	 * @param script
	 * @return sha值
	 */
	@Override
	public String scriptLoad(String script) {
		return redisTemplate.execute(new RedisCallback<String>() {
			public String doInRedis(RedisConnection connection) {
				return connection.scriptLoad(getStringSerialize(script));
			}
		});
	}

	@Override
	public String lPop(String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) {
				return getStringDeserialize(connection.lPop(getStringSerialize(key)));
			}
		});
	}
	@Override
	public String rPop(String key) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) {
				return getStringDeserialize(connection.rPop(getStringSerialize(key)));
			}
		});
	}
	@Override
	public void lTrim(String key, Long begin, Long end) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) {
				connection.lTrim(getStringSerialize(key), begin, end);
				return null;
			}
		});
	}
	@Override
	public Set<String> keys(String patternKey) {
		return redisTemplate.execute(new RedisCallback<Set<String>>() {
			@Override
			public Set<String> doInRedis(RedisConnection connection) {
				Set<byte[]> setC = connection.keys(getStringSerialize(patternKey));
				Set<String> setT = new HashSet<String>();
				if (setC != null) {
					for (byte[] bs : setC) {
						setT.add(getStringDeserialize(bs));
					}
				}
				return setT;
			}
		});
	}

	@Override
	public void pSubscribe(RedisMsgListener redisMsgListener, String... pMsgKeys) {
		if (pMsgKeys == null) {
			logger.error("redis patternKeys is null");
			return;
		}
		byte[][] key = new byte[pMsgKeys.length][];
		for (int i = 0; i < pMsgKeys.length; i++) {
			String string = pMsgKeys[i];
			if (string == null) {
				logger.error("redis patternKey[i] is null");
				return;
			}
			key[i] = getStringSerialize(string);
		}
		if (key.length <= 0) {
			return;
		}
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) {
				connection.pSubscribe(new MessageListener() {
					@Override
					public void onMessage(Message message, byte[] pattern) {
						redisMsgListener.onMessage(getStringDeserialize(pattern),
								getStringDeserialize(message.getChannel()), getStringDeserialize(message.getBody()));
					}
				}, key);
				return null;
			}
		});
	}

	@Override
	public Long publish(String msgKey, String msg) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.publish(getStringSerialize(msgKey), getStringSerialize(msg));
			}
		});
	}

	private static final String SetByLockScript = "" + "local oldValue = redis.call('GET', KEYS[1]);\n"
			+ "if oldValue and type(oldValue)=='string' and oldValue==KEYS[3] then\n"
			+ "local ttls = redis.call('TTL', KEYS[1]); \n" + "redis.call('SET',KEYS[1], KEYS[2]);\n"
			+ "if ttls and ttls>0 then\n" + "redis.call('EXPIRE',KEYS[1],ttls);\n" + "end\n"
			+ "return tostring(true);\n" + "end\n" + "return tostring(false)";
	private static final DefaultRedisScript<String> setByLockScript = new DefaultRedisScript<String>(SetByLockScript,
			String.class);

	@Override
	public boolean setByLock(String key, String newValue, String oldValue) {
		try {
			String res = eval(setByLockScript.getScriptAsString(), String.class, key, newValue, oldValue);
			if (Boolean.TRUE.toString().equals(res)) {
				return true;
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public <T> T getCash(RedisCash<T> cash) {
		if (cash == null) {
			return null;
		}
		String rediskey = cash.getRediskey();
		String k = get(rediskey);
		if (k == null || k.isEmpty()) {
			T t = cash.useCash();
			if (t != null) {
				set(rediskey, JsonUtil.toJSONString(t));
				if (cash.getSeconds() != null && cash.getSeconds() > 0) {
					expire(rediskey, cash.getSeconds());
				}
				return t;
			} else {
				return null;
			}
		} else {
			T res = JsonUtil.parseObject(k, cash.getClazz());
			return res;
		}
	}

	private static final String LPopBakScript = "" + "local tt = redis.call('rpop', KEYS[1]);\n" + "if tt then\n"
			+ "redis.call('hset', KEYS[2], KEYS[3], tt);\n" + "return tt;\n" + "end\n" + "return nil";;
	private static final DefaultRedisScript<String> lPopBakScript = new DefaultRedisScript<String>(LPopBakScript,
			String.class);

	@Override
	public Boolean lPopBak(String key, LPopBakAction action) {
		String bak_keyname = key + ":bak"; // map
		String mapkey_keyval = UUID.randomUUID().toString();
		String val = eval(lPopBakScript.getScriptAsString(), String.class, key, bak_keyname, mapkey_keyval);
		try {
			String res = action.exec(val);
			if (res == null || res.isEmpty()) {
				hDel(bak_keyname, mapkey_keyval);
				return true;
			} else {
				String error_keyname = key + ":error"; // map
				hSet(error_keyname, mapkey_keyval, res);
			}
		} catch (Exception e) {
			String error_keyname = key + ":error"; // map
			hSet(error_keyname, mapkey_keyval, e.getMessage());
		}
		return false;
	}

	@Override
	public boolean exists(String key) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.exists(getStringSerialize(key));
			}
		});
	}
	
	@Override
	public boolean exists(byte[] key) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.exists(key);
			}
		});
	}

	@Override
	public String hGet(String key, String field) {
		return redisTemplate.execute(new RedisCallback<String>() {
			@Override
			public String doInRedis(RedisConnection connection) {
				return getStringDeserialize(connection.hGet(getStringSerialize(key), getStringSerialize(field)));
			}
		});
	}
	
	@Override
	public Map<String, String> hGetAll(String key) {
		Map<byte[], byte[]> map = redisTemplate.execute(new RedisCallback<Map<byte[], byte[]>>() {
			@Override
			public Map<byte[], byte[]> doInRedis(RedisConnection connection) {
				return connection.hGetAll(getStringSerialize(key));
			}
		});
		return getStringDeserialize(map);
	}

	@Override
	public List<String> lRange(String key, Long begin, Long end) {
		List<byte[]> bList = redisTemplate.execute(new RedisCallback<List<byte[]>>() {
			@Override
			public List<byte[]> doInRedis(RedisConnection connection) {
				return connection.lRange(getStringSerialize(key), begin, end);
			}
		});
		return getStringDeserialize(bList);
	}
	
	@Override
	public List<String> hMGet(String key, String... fields) {
		List<byte[]> bList = redisTemplate.execute(new RedisCallback<List<byte[]>>() {
			@Override
			public List<byte[]> doInRedis(RedisConnection connection) {
				return connection.hMGet(getStringSerialize(key), getStringSerializes(fields));
			}
		});
		return getStringDeserialize(bList);
	}

	@Override
	public Long lRem(String key, Long count, String value) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.lRem(getStringSerialize(key), count, getStringSerialize(value));
			}
		});
	}
	@Override
	public Long incr(String key){
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.incr(getStringSerialize(key));
			}
		});
	}
	@Override
	public Long hIncrBy(String key, String field, Long delta){
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.hIncrBy(getStringSerialize(key), getStringSerialize(field), delta);
			}
		});
	}
	
	@Override
	public byte[] hGet(byte[] key, byte[] field) {
		return redisTemplate.execute(new RedisCallback<byte[]>() {
			@Override
			public byte[] doInRedis(RedisConnection connection) {
				return connection.hGet(key, field);
			}
		});
	}
	
	@Override
	public Long hLen(byte[] key) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.hLen(key);
			}
		});
	}
	
	@Override
	public Boolean hExists(byte[] key, byte[] field) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.hExists(key, field);
			}
		});
	}
	
	@Override
	public Boolean hSet(byte[] key, byte[] field,byte[] value) {
		return redisTemplate.execute(new RedisCallback<Boolean>() {
			@Override
			public Boolean doInRedis(RedisConnection connection) {
				return connection.hSet(key, field, value);
			}
		});
	}
	
	@Override
	public Long hDel(byte[] key, byte[]... fields) {
		return redisTemplate.execute(new RedisCallback<Long>() {
			@Override
			public Long doInRedis(RedisConnection connection) {
				return connection.hDel(key, fields);
			}
		});
	}
	
	@Override
	public Set<byte[]> hKeys(byte[] key) {
		return redisTemplate.execute(new RedisCallback<Set<byte[]>>() {
			@Override
			public Set<byte[]> doInRedis(RedisConnection connection) {
				return connection.hKeys(key);
			}
		});
	}
	
	@Override
	public List<byte[]> hVals(byte[] key) {
		return redisTemplate.execute(new RedisCallback<List<byte[]>>() {
			@Override
			public List<byte[]> doInRedis(RedisConnection connection) {
				return connection.hVals(key);
			}
		});
	}
	
	@Override
	public void setEx(String key,long seconds,String value) {
		setEx(getStringSerialize(key), seconds, getStringSerialize(value));
	}
	
	@Override
	public void setEx(byte[] key,long seconds,byte[] value) {
		redisTemplate.execute(new RedisCallback<Object>() {
			@Override
			public Object doInRedis(RedisConnection connection) {
				connection.setEx(key, seconds, value);
				return null;
			}
		});
	}
}
